package Thread::EventBus;

=head1 NAME

Thread::EventBus - an event-bus to broker requests threads using the autoloader

=head1 VERSION

1.0

=head1 SYNPOSIS

  package WorkerThread;
  use threads;
  use threads::shared;
  use Thread::EventBus;
  
  sub new
  {
      my $class = shift;
      my $id = shift;
      my $self = {};
      bless $self, $class;
      
      $self->{ 'eb' } = Thread::EventBus->new( $self, 'worker#' . $id );
      $self->{ 'eb' }->set_procedure( 'method1', \&method1 );
      $self->{ 'eb' }->set_method( 'method2', \&method2 );
      $self->{ 'id' } = $id;
      
      $self->thread_start;
      return $self;
  }
  
  sub thread_start
  {
      my $self = shift;
      
      threads->create( \&threadloop, $self );
      return;
  }
  
  sub threadloop
  {
      my $self = shift;
      $self->{ 'eb' }->run;
  }

  sub method1
  {
      my $self = shift;
      my $data = shift;
      
      warn "Hello from method1(), thread $self->{id}!  Parameter: $data\n";
      return;
  }  
  
  sub method2
  {
      my $self = shift;
      my $data = shift;
      warn "Hello from method2(), thread $self->{id}!  Parameter: $data Returning the thread ID.\n";
      return $self->{ 'id' };
  }
  
  1;
  
  my $workerthread = WorkerThread->new( 1 );
  my $workerthread2 = WorkerThread->new( 2 );
  
  $workerthread->{eb}->method1( 'foo bar' );
  print "Result from method2: " . $workerthread2->{eb}->method2( 'baz' ) . "\n";

  sleep 1;
    
  
=head1 DESCRIPTION

Thread::EventBus is most useful if used in a threaded system which wants to
expose a Perl object as a worker thread.  The model is to wait for calls on
a Thread::Queue.  Calls to the thread are made by a single shared object
which automatically serialises parameters made by autoloader calls.

Calls can either represent methods or procedures.  A method will block the
caller until the function returns, and will shared clone the return as
necessary.  A procedure will simply start the call, and return to the
calling thread.

=head1 USAGE

=over 4

=item new( $object, $id )

Instantiates new EventBus.  $object is the value of the $self passed to the
autoloaded functions.  $id is an arbitrary scalar to define your thread
context.

=item set_procedure( $name, \&fn )

Configures the EventBus with a new procedure.  When the $name autoload
method is invoked, \&fn is called, using the $object set in new() as self.

Procedure return values are ignored, and the caller thread does not block
while waiting for \&fn to finish.

=item set_method( $name, \&fn )

Configures the EventBus with a new method.  When the $name autoload
method is invoked, \&fn is called, using the $object set in new() as self.

Return value is thread cloned as needed, and the caller thread blocks until
\&fn is finished.  The return value is returned to it.

=item run()

This starts the EventBus loop which listens in on a Thread::Queue for calls. 
You're going to need this.

=head1 HOMEPAGE

http://code.google.com/p/threadeventbus/

=head1 AUTHORS

Written by Matt Busigin <mbusigin@gmail.com>

=head1 LICENSE

Copyright (c) 2010, Matt Busigin
All rights reserved.

Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:

Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.
Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.
Neither the name of Matt Busigin nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

=head1 DISCLAIMER

THIS SOFTWARE IS PROVIDED `AS IS'' AND ANY EXPRESS OR IMPLIED
WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY DIRECT, INDIRECT,
INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS
OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR
TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH
DAMAGE.

In other words: Use at your own risk.  Provided as is.  Your mileage
may vary.  Read the source, Luke!

And finally, just to be sure:

Any Use of This Product, in Any Manner Whatsoever, Will Increase the
Amount of Disorder in the Universe. Although No Liability Is Implied
Herein, the Consumer Is Warned That This Process Will Ultimately Lead
to the Heat Death of the Universe.

=cut

use strict;
use AutoLoader;
use Data::Dumper;
use threads;
use threads::shared;
use Thread::Queue;
use Clone qw(clone);
use Thread::Semaphore;


our $AUTOLOAD;


sub new
{
    my $class = shift;
    my $self_self = shift;
    my $name = shift;

    my $txqueue: shared = Thread::Queue->new() or die "Couldn't create thread queue\n";
    my $rxqueue: shared = Thread::Queue->new() or die "Couldn't create thread queue\n";
    my $s :shared = Thread::Semaphore->new();

    my $self = 
    {
        'name'      => $name,
        'rxqueue'       => $rxqueue,
        'txqueue'       => $txqueue,
        'semaphore' => $s,
        'key'           => 'value',
        'methods'       => 
                        {
                                'test'          => \&footest,
                        },
            'procedures' =>
                {
                    'test'      => \&footest,
                },
                'self'      => $self_self,
                'pre_call_trap' => undef,
                'post_call_trap' => undef,
    };
    bless $self, $class;
    
    return $self;
}


sub AUTOLOAD
{
        my $self = shift;
        my $type = ref($self)
                    or die "$self is not an object";

        my $name = $AUTOLOAD;
        $name =~ s/.*://;   # strip fully-qualified portion
        return $self->call_thread( $name, @_ );
}


sub set_procedure
{
    my ( $self, $name, $fn ) = @_;
    $self->{ 'procedures' }->{ $name } = $fn;
    return;
}


sub set_method
{
    my ( $self, $name, $fn ) = @_;
    $self->{ 'methods' }->{ $name } = $fn;
    return;
}


sub call_thread
{
    my $self = shift;
    my $name = shift;
    my $args = \@_;
 
#    print "call_thread(): $self->{name} :: $name:\n" . Dumper( $args ) . "\n\n";
#    warn "$self->{name} :: $name Semaphore down\n";
#    $self->{'semaphore'}->down(); 
#    warn "\t$self->{name} :: $name Enqueue & clone\n";
    my $txqueue: shared = Thread::Queue->new() or die "Couldn't create thread queue\n";
    my @send = threads::shared::shared_clone([$name, $args, $txqueue]);
    $self->{rxqueue}->enqueue( @send );
    goto call_thread_done if ( $self->{'procedures'}->{$name} );
    
#    warn "\t$self->{name} :: $name Dequeue reply\n";
    my $ret = $txqueue->dequeue;
#    warn "\t$self->{name} :: $name Semaphore up\n";
#    $self->{'semaphore'}->up(); 

call_thread_done:
#    warn "$self->{name} :: $name Done\n";
    return $ret if defined($ret);
}

sub set_self
{
    my $self = shift;
    my $newself = shift;
    $self->{'self'} = shift;
}

sub run
{
    my $self = shift;
    
    while( 1 )
    {
        $self->oneiteration;
    }
}


sub oneiteration
{
        my $self = shift;
        
        my $message = $self->{rxqueue}->dequeue;
        {
                if ( scalar(@{$message}) != 3 )
                {
                    print "\n\nW T F: message is not 3:\n" . Dumper( $message ) . "\n\n";
                    next;
                }
                my $command = $message->[0];
                my $args = $message->[1];
                my $txqueue = $message->[2];
                
                my $procedure = 0;
                my $func = $self->{methods}->{$command};
                if ( !$func )
                {
                    $func = $self->{procedures}->{$command};
                    $procedure = 1;
                    if ( !$func )
                    {
                            die "No such function or procedure '$command'\n" . Dumper( $self->{'methods'} ) . Dumper( $self->{'procedures'} );
                    }
                }

                my @arguments;
                push @arguments, $self->{'self'};
                if ( ref($args) eq 'ARRAY')
                {
                        foreach my $p ( @{$args} )
                        {
                                push @arguments, $p;
                        }
                }
                else
                {
                        push @arguments, $args;
                }


                $self->{'pre_call_trap'}->( \@arguments )
                    if ( $self->{'pre_call_trap'} );

                my $ret = $func->( @arguments ); #@{ $self, $args } );
#               warn "      Return for $command : " . Dumper( $ret );
                $self->{'post_call_trap'}->( $self, \@arguments, $ret )
                    if ( $self->{'post_call_trap'} );
                if ( $procedure != 1 )
                {
                    $ret = 0 if !defined($ret);
                    $txqueue->enqueue( threads::shared::shared_clone($ret) );
                }
        }
}

sub footest
{
        my $self = shift;
        my ( $a, $b, $c ) = @_;
        return $a + $b + $c;
}


1;
